package com.atguigu.gmall.realtime.function;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.bean.TableProcess;
import com.atguigu.gmall.realtime.common.GmallConfig;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.IOUtils;
import org.apache.flink.util.OutputTag;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.util.Arrays;

/**
 * 继承广播流处理函数，实现动态分流
 * <p>
 * 说明：<br>
 * 1、将维度数据放到维度侧输出流 <br>
 * 2、事实数据放到主流
 *
 * @author lvbingbing
 * @date 2022-03-19 16:35
 */
@Slf4j
public class DynamicSplitStreamBroadcastProcessor extends BroadcastProcessFunction<JSONObject, String, JSONObject> {

    private static final long serialVersionUID = 2822636931834242052L;
    /**
     * 维度侧输出流的标记
     */
    private OutputTag<JSONObject> dimTag;

    /**
     * 广播状态描述器
     */
    private MapStateDescriptor<String, TableProcess> mapStateDescriptor;

    /**
     * 数据库连接对象
     */
    private transient Connection connection;

    @SuppressWarnings("unused")
    public DynamicSplitStreamBroadcastProcessor() {
    }

    public DynamicSplitStreamBroadcastProcessor(OutputTag<JSONObject> dimTag, MapStateDescriptor<String, TableProcess> mapStateDescriptor) {
        this.dimTag = dimTag;
        this.mapStateDescriptor = mapStateDescriptor;
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        Class.forName("org.apache.phoenix.jdbc.PhoenixDriver");
        connection = DriverManager.getConnection(GmallConfig.PHOENIX_SERVER);
    }

    @Override
    public void processElement(JSONObject jsonObject, BroadcastProcessFunction<JSONObject, String, JSONObject>.ReadOnlyContext ctx, Collector<JSONObject> out) throws Exception {
        // 处理业务流中的数据，Maxwell 从业务数据库中采集到的数据
        String table = jsonObject.getString("table");
        String type = jsonObject.getString("type");
        // 在使用 Maxwell 处理历史数据时，类型是 bootstrap-insert
        if (type.equals("bootstrap-insert")) {
            type = "insert";
            jsonObject.put("type", type);
        }
        // 获取广播状态
        ReadOnlyBroadcastState<String, TableProcess> broadcastState = ctx.getBroadcastState(mapStateDescriptor);
        String key = table + ":" + type;
        // 从状态中获取配置信息
        TableProcess tableProcess = broadcastState.get(key);
        // 如果找到了配置信息，判断是事实数据还是维度数据
        if (tableProcess != null) {
            // 不管是维度数据，还是事实数据，都要设置输出表
            String sinkTable = tableProcess.getSinkTable();
            jsonObject.put("sink_table", sinkTable);
            // 向下游传递数据之前，将不需要的字段过滤掉
            filterColumns(jsonObject.getJSONObject("data"), tableProcess.getSinkColumns());
            String sinkType = tableProcess.getSinkType();
            // 如果是维度数据
            if (TableProcess.SINK_TYPE_HBASE.equals(sinkType)) {
                ctx.output(dimTag, jsonObject);
            } else if (TableProcess.SINK_TYPE_KAFKA.equals(sinkType)) {
                out.collect(jsonObject);
            }
        } else {
            // 在配置表中没有该操作对应的配置
            log.info("No this Key in TableProcess：{}", key);
        }
    }

    /**
     * 将不需要的字段过滤掉
     *
     * @param jsonDataObject 原始数据。格式示例：{"tm_name":"aaa","logo_url":"aa","id":12}
     * @param sinkColumns    配置表中配置的要输出的字段信息。格式举例：id,tm_name
     */
    private void filterColumns(JSONObject jsonDataObject, String sinkColumns) {
        String[] columnNames = sinkColumns.split(",");
        jsonDataObject.entrySet()
                .removeIf(entry -> Arrays.stream(columnNames)
                        .noneMatch(s -> s.equals(entry.getKey())));
    }

    @Override
    public void processBroadcastElement(String jsonString, BroadcastProcessFunction<JSONObject, String, JSONObject>.Context ctx, Collector<JSONObject> out) throws Exception {
        // 处理广播流中的数据，FlinkCDC 从 MySQL 中读取配置表
        // 1. 将 JSON 字符串转换成 JSON对象
        JSONObject jsonObject = JSON.parseObject(jsonString);
        // 2. 将 JSON 对象转换成 Java 对象
        TableProcess tableProcess = JSON.toJavaObject(jsonObject.getJSONObject("data"), TableProcess.class);
        // 获取操作类型(insert、update、delete)
        String operateType = tableProcess.getOperateType();
        // 获取来源表
        String sourceTable = tableProcess.getSourceTable();
        // 拼接 key
        String key = sourceTable + ":" + operateType;
        // 3. 获取广播状态，将配置信息放到广播状态中
        BroadcastState<String, TableProcess> broadcastState = ctx.getBroadcastState(mapStateDescriptor);
        broadcastState.put(key, tableProcess);
        // 4、如果HBase中不存在维度表，创建维度表
        String sinkType = tableProcess.getSinkType();
        if (sinkType.equals(TableProcess.SINK_TYPE_HBASE)) {
            createDimTableWhereNotExist(tableProcess);
        }
    }

    /**
     * 如果 HBase 中不存在维度表，创建维度表
     *
     * @param tableProcess 配置表对象
     */
    private void createDimTableWhereNotExist(TableProcess tableProcess) {
        // 1、空值处理
        String sinkPk = tableProcess.getSinkPk();
        if (sinkPk == null) {
            sinkPk = "id";
        }
        String sinkExtend = tableProcess.getSinkExtend();
        if (sinkExtend == null) {
            sinkExtend = StringUtils.EMPTY;
        }
        // 2、拼接建表 sql
        String tableName = tableProcess.getSinkTable();
        StringBuilder stringBuilder = new StringBuilder();
        stringBuilder.append("create table if not exists ")
                .append(GmallConfig.HBASE_SCHEMA)
                .append(".")
                .append(tableName)
                .append(" (");
        String sinkColumns = tableProcess.getSinkColumns();
        String[] columnNames = sinkColumns.split(",");
        for (String columnName : columnNames) {
            stringBuilder.append(columnName);
            if (columnName.equals(sinkPk)) {
                stringBuilder.append(" varchar primary key ");
            } else {
                stringBuilder.append(" varchar ");
            }
            stringBuilder.append(",");
        }
        int length = stringBuilder.length();
        String sql = stringBuilder.replace(length - 1, length, ") ")
                .append(sinkExtend)
                .toString();
        log.info("phoenix 中的建表语句：{}", sql);
        // 3、执行 sql
        PreparedStatement ps = null;
        try {
            ps = connection.prepareStatement(sql);
            ps.execute();
        } catch (Exception e) {
            log.error("在 phoenix 中建表失败！原因：{}", e);
            throw new RuntimeException("在 phoenix 中建表失败！");
        } finally {
            IOUtils.closeQuietly(ps);
        }
    }

    @Override
    public void close() throws Exception {
        IOUtils.closeQuietly(connection);
    }
}